//
// Created by 25429 on 2020/6/8.
//

#include "trade_engine.h"
#include "strategy_engine.h"
#include <utility>

#pragma clang diagnostic push
#pragma ide diagnostic ignored "EndlessLoop"

#include "oes_client_my_spi_sample.h"

void trade_engine::go() {
    engine_base::go();

    m_api = new Quant360::OesClientApi();
    m_spi = new OesClientMySpi();

    new std::thread([&]() {
        // 绑定CPU
        common_function::bind_cpu(m_log_title);
        m_spi->SetTradeEngine(this);

        /* 打印API版本信息 */
        LOG4CPLUS_INFO(MY_LOGGER, m_log_title << "[OesClientApi version]" << Quant360::OesClientApi::GetVersion());
        /* 注册spi回调接口 */
        m_api->RegisterSpi((Quant360::OesClientSpi *) m_spi);

        /* 加载配置文件 */
        auto config_path = m_config.m_params[TRADE_CONFIG_PATH].toString();
        // 设置硬盘序列号
        m_api->SetCustomizedDriverId(m_config.m_params[TRADE_DRIVER_ID].toString().toStdString().c_str());

        /* 启动 */
        int re_start_times = 3;
        while (true) {
            if (!m_api->LoadCfg(config_path.toStdString().c_str())) {
                LOG4CPLUS_ERROR(MY_LOGGER, m_log_title << config_path.toStdString() << "load config failed");
                exit(EXIT_FAILURE);
            }

            if (!m_api->Start()) {
                LOG4CPLUS_ERROR(MY_LOGGER, m_log_title << "api start failed");
                sleep(1);
                re_start_times--;
                if (re_start_times == 0) {
                    LOG4CPLUS_INFO(MY_LOGGER, m_log_title << "api restart failed");
                    exit(EXIT_FAILURE);
                }
                continue;
            }
            break;
        }

        // 查询持仓
        req_trade_holdings();
        req_trade_balances();

        // 通知策略交易初始化成功
        SiriusQueueFastField data_fast[ITEM_SIZE_1024];
#pragma clang diagnostic push
#pragma clang diagnostic ignored "-Wmissing-noreturn"
        while (true) {
            auto size = m_trade_queue_fast.try_dequeue_bulk(data_fast, ITEM_SIZE_1024);
            if (size > 0) {
                for (auto i = 0; i < size; ++i) {
                    switch (data_fast[i].type) {
                        case SiriusQueueType::QUEUE_TRADE_ORDER_CANCEL: {
                            std::lock_guard<std::mutex> mtx_locker(m_api_mtx);

                            auto &order = m_spi->m_system_order_no_insert[data_fast[i].order.system_order_no];
                            if (order.order_id != -1) {
                                OesOrdCancelReqT cancelReq = {NULLOBJ_OES_ORD_CANCEL_REQ};
                                cancelReq.origClOrdId = order.order_id;
                                cancelReq.mktId = order.market;
                                cancelReq.clSeqNo = static_cast<int32>(++m_api->apiEnv.ordChannel.lastOutMsgSeq);
                                auto ret = m_api->SendCancelOrder(&cancelReq);
                                if (ret < 0) {
                                    util::Singleton<my_log>::Instance()->add_log(
                                            m_log_title + "api cancel order failed:" + to_string(ret));
                                }

                                util::Singleton<time_cost>::Instance()->add_cost(data_fast[i].order.original_point,
                                                                                 time_cost::type::order_cancel_t);
                            } else {
                                order.need_cancel = true;
                            }
                            util::Singleton<my_log>::Instance()->add_log(m_log_title
                                                                         + "order cancel, [symbol]"
                                                                         + order.symbol
                                                                         + "[vol]"
                                                                         + to_string(order.volume)
                                                                         + "[system_order_no]"
                                                                         + to_string(data_fast[i].order.system_order_no)
                                                                         + "[order_id]"
                                                                         + to_string(order.order_id));
                        }
                            break;
                        case SiriusQueueType::QUEUE_TRADE_ORDER_INSERT: {
                            std::lock_guard<std::mutex> mtx_locker(m_api_mtx);

                            auto market = data_fast[i].order.symbol.substr(0, 2) == "11" ? OES_MKT_SH_ASHARE
                                                                                         : OES_MKT_SZ_ASHARE;

                            OesOrdReqT ordReq = {NULLOBJ_OES_ORD_REQ};
                            ordReq.clSeqNo = (int32) ++m_api->apiEnv.ordChannel.lastOutMsgSeq;
                            m_spi->m_order_reference_system_order_no[common_function::env_client_seq_no(
                                    m_api->apiEnv.ordChannel.__clEnvId,
                                    ordReq.clSeqNo)] = data_fast[i].order.system_order_no;
                            ordReq.mktId = market;
                            ordReq.ordType = OES_ORD_TYPE_LMT;
                            ordReq.bsType = data_fast[i].order.direction == TRADE_DIRECTION_SELL ? OES_BS_TYPE_SELL
                                                                                                 : OES_BS_TYPE_BUY;

                            strncpy(ordReq.securityId, data_fast[i].order.symbol.c_str(),
                                    sizeof(ordReq.securityId) - 1);

                            ordReq.ordQty = data_fast[i].order.volume;
                            ordReq.ordPrice = data_fast[i].order.price;

                            data_fast[i].order.market = market;

                            m_spi->m_system_order_no_insert[data_fast[i].order.system_order_no] = data_fast[i].order;

                            auto ret = m_api->SendOrder(&ordReq);
                            if (ret < 0) {
                                util::Singleton<my_log>::Instance()->add_log(
                                        m_log_title + "api send order failed:" + to_string(ret));
                            }

                            util::Singleton<time_cost>::Instance()->add_cost(data_fast[i].order.original_point,
                                                                             time_cost::type::order_send_t);

                            util::Singleton<my_log>::Instance()->add_log(m_log_title + "send order, "
                                                                         + "[symbol]"
                                                                         + data_fast[i].order.symbol
                                                                         + "[system_order_no]"
                                                                         +
                                                                         to_string(data_fast[i].order.system_order_no)
                                                                         + "[direction]"
                                                                         + to_string(data_fast[i].order.direction));
                        }
                            break;
                        default:
                            break;
                    }
                }
            }
        }
#pragma clang diagnostic pop
    });
    new std::thread([&]() {
        common_function::bind_cpu(m_log_title);
        // 持仓
        vector<SiriusTradeRspHoldingField> trade_holdings{};
        SiriusTradeRspBalanceField trade_balance;
        // 通知策略交易初始化成功
        SiriusQueueField data_slow[ITEM_SIZE_1024];
#pragma clang diagnostic push
#pragma clang diagnostic ignored "-Wmissing-noreturn"
        while (true) {
            auto size = m_trade_queue_slow.try_dequeue_bulk(data_slow, ITEM_SIZE_1024);
            if (size > 0) {
                for (auto i = 0; i < size; ++i) {
                    switch (data_slow[i].type) {
                        case SiriusQueueType::QUEUE_TRADE_ORDER_REJECTED:
                            if (m_spi->m_system_order_no_insert.find(
                                    data_slow[i].trade_order_rejected.system_order_no) ==
                                m_spi->m_system_order_no_insert.end())
                                break;
                            util::Singleton<my_log>::Instance()->add_log(m_log_title + "order rejected, [symbol]"
                                                                         +
                                                                         m_spi->m_system_order_no_insert[data_slow[i].trade_order_rejected.system_order_no].symbol
                                                                         + "[system_order_no]"
                                                                         + to_string(
                                    data_slow[i].trade_order_rejected.system_order_no)
                                                                         + "[error_id]"
                                                                         + to_string(
                                    data_slow[i].trade_order_rejected.error_id));
                            break;
                        case SiriusQueueType::QUEUE_TRADE_ORDER_RECEIVED: {
                            if (m_spi->m_system_order_no_insert.find(
                                    data_slow[i].trade_order_received.system_order_no) ==
                                m_spi->m_system_order_no_insert.end()) {
                                break;
                            }

                            std::lock_guard<std::mutex> mtx_locker(m_api_mtx);

                            m_spi->m_system_order_no_insert[data_slow[i].trade_order_received.system_order_no].order_id = data_slow[i].trade_order_received.order_id;
                            if (m_spi->m_system_order_no_insert[data_slow[i].trade_order_received.system_order_no].need_cancel) {
                                OesOrdCancelReqT cancelReq = {NULLOBJ_OES_ORD_CANCEL_REQ};
                                cancelReq.origClOrdId = data_slow[i].trade_order_received.order_id;
                                cancelReq.mktId = m_spi->m_system_order_no_insert[data_slow[i].trade_order_received.system_order_no].market;
                                cancelReq.clSeqNo = static_cast<int32>(++m_api->apiEnv.ordChannel.lastOutMsgSeq);
                                auto ret = m_api->SendCancelOrder(&cancelReq);
                                if (ret < 0) {
                                    util::Singleton<my_log>::Instance()->add_log(
                                            m_log_title + "api cancel order failed:" + to_string(ret));
                                } else {
                                    util::Singleton<my_log>::Instance()->add_log(
                                            m_log_title + "api cancel order success:" + to_string(ret));
                                }
                            }

                            util::Singleton<my_log>::Instance()->add_log(m_log_title + "order received, [symbol]"
                                                                         +
                                                                         m_spi->m_system_order_no_insert[data_slow[i].trade_order_received.system_order_no].symbol
                                                                         + "[system_order_no]"
                                                                         + to_string(
                                    data_slow[i].trade_order_received.system_order_no)
                                                                         + "[order_id]"
                                                                         + to_string(
                                    data_slow[i].trade_order_received.order_id));
                        }

                            break;
                        case SiriusQueueType::QUEUE_TRADE_ORDER_UPDATE:
                            if (m_spi->m_system_order_no_insert.find(
                                    data_slow[i].trade_order_update.system_order_no) ==
                                m_spi->m_system_order_no_insert.end())
                                break;
                            util::Singleton<my_log>::Instance()->add_log(m_log_title + "order update, [symbol]"
                                                                         +
                                                                         m_spi->m_system_order_no_insert[data_slow[i].trade_order_update.system_order_no].symbol
                                                                         + "[system_order_no]"
                                                                         + to_string(
                                    data_slow[i].trade_order_update.system_order_no)
                                                                         + "[order_id]" +
                                                                         to_string(
                                                                                 data_slow[i].trade_order_update.order_id)
                                                                         + "[status]"
                                                                         + order_status2(
                                    data_slow[i].trade_order_update.order_status));
                            break;
                        case SiriusQueueType::QUEUE_TRADE_ORDER_DEAL: {
                            auto it_order = m_spi->m_system_order_no_insert.find(
                                    data_slow[i].trade_order_deal.system_order_no);
                            if (it_order == m_spi->m_system_order_no_insert.end()) {
                                break;
                            }
                            util::Singleton<my_log>::Instance()->add_log(m_log_title + "order deal, [symbol]"
                                                                         +
                                                                         m_spi->m_system_order_no_insert[data_slow[i].trade_order_deal.system_order_no].symbol
                                                                         + "[system_order_no]"
                                                                         + to_string(
                                    data_slow[i].trade_order_deal.system_order_no)
                                                                         + "[deal_volume]"
                                                                         + to_string(
                                    data_slow[i].trade_order_deal.deal_volume));
                        }
                            break;
                        case SiriusQueueType::QUEUE_TRADE_ORDER_CANCEL_RECEIVED:
                            if (m_spi->m_system_order_no_insert.find(
                                    data_slow[i].trade_order_cancel_received.system_order_no) ==
                                m_spi->m_system_order_no_insert.end()) {
                                break;
                            }
                            util::Singleton<my_log>::Instance()->add_log(m_log_title + "cancel received, [symbol]"
                                                                         +
                                                                         m_spi->m_system_order_no_insert[data_slow[i].trade_order_cancel_received.system_order_no].symbol
                                                                         + "[system_order_no]"
                                                                         + to_string(
                                    data_slow[i].trade_order_cancel_received.system_order_no));
                            break;
                        case SiriusQueueType::QUEUE_TRADE_ORDER_CANCEL_REJECTED:
                            if (m_spi->m_system_order_no_insert.find(
                                    data_slow[i].trade_order_cancel_rejected.system_order_no) ==
                                m_spi->m_system_order_no_insert.end()) {
                                break;
                            }
                            util::Singleton<my_log>::Instance()->add_log(m_log_title + "order cancel rejected, [symbol]"
                                                                         +
                                                                         m_spi->m_system_order_no_insert[data_slow[i].trade_order_cancel_rejected.system_order_no].symbol
                                                                         + "[system_order_no]"
                                                                         + to_string(
                                    data_slow[i].trade_order_cancel_rejected.system_order_no)
                                                                         + "[error_id]"
                                                                         + to_string(
                                    data_slow[i].trade_order_cancel_rejected.error_id));

                            break;
                        case SiriusQueueType::QUEUE_TRADE_BALANCES: {
                            std::lock_guard<std::mutex> mtx_locker(m_api_mtx);
                            m_api->QueryCashAsset(nullptr, 0);
                        }
                            break;
                        case SiriusQueueType::QUEUE_TRADE_HOLDING: {
                            std::lock_guard<std::mutex> mtx_locker(m_api_mtx);
                            m_api->QueryStkHolding(nullptr, 0);
                            new std::thread([&]() {
                                sleep(2);
                                if (trade_holdings.empty()) {
                                    m_strategy_engine->m_queue.enqueue(SiriusQueueField(trade_holdings));
                                }
                            });
                        }
                            break;
                        case SiriusQueueType::QUEUE_TRADE_HOLDING_RSP: {
                            trade_holdings = data_slow[i].trade_holdings;
                            m_strategy_engine->m_queue.enqueue(SiriusQueueField(data_slow[i].trade_holdings));
                            int64 holding_balance = 0;
                            for (auto &it: trade_holdings) {
                                holding_balance += it.sumHld * it.costPrice;
                            }
                            util::Singleton<my_log>::Instance()->add_log(
                                    m_log_title + "output_holding_balance " + to_string(holding_balance));
                        }
                            break;
                        case SiriusQueueType::QUEUE_TRADE_BALANCES_RSP:
                            trade_balance = data_slow[i].trade_balance;
                            util::Singleton<my_log>::Instance()->add_log(
                                    m_log_title + "output_trade_balance " + to_string(trade_balance.currentTotalBal));
                            break;
                        case SiriusQueueType::QUEUE_TRADE_ORDERS: {
                            std::lock_guard<std::mutex> mtx_locker(m_api_mtx);
                            m_api->QueryOrder(nullptr, 0);
                        }
                            break;
                        case SiriusQueueType::QUEUE_TRADE_DEALS: {
                            std::lock_guard<std::mutex> mtx_locker(m_api_mtx);
                            m_api->QueryTrade(nullptr, 0);
                        }
                            break;
                        default:
                            break;
                    }
                }
            }
        }
#pragma clang diagnostic pop
    });
}

int trade_engine::order_status(int status) {
    if (status == 8) return SIRIUS_ORDER_STATUS_FILLED;
    if (status == 3) return SIRIUS_ORDER_STATUS_PART_FILLED;
    if (status == 6) return SIRIUS_ORDER_STATUS_PART_FILLED_WITHDRAW;
    if (status == 2) return SIRIUS_ORDER_STATUS_STOCK_QUEUE;
    if (status == 5) return SIRIUS_ORDER_STATUS_STOCK_SENT_CANCELING;
    if (status == 7) return SIRIUS_ORDER_STATUS_CANCELED;
    if (status == 0) return SIRIUS_ORDER_STATUS_UN_KNOW;

    return SIRIUS_ORDER_STATUS_UN_KNOW;
}

string trade_engine::order_status2(int status) {
    switch (status) {
        case 0:
            return "unknown";
        case 2:
            return "queue";
        case 3:
            return "part filled";
        case 5:
            return "canceling";
        case 6:
            return "part filled part withdraw";
        case 7:
            return "canceled";
        case 8:
            return "filled";
        default:
            break;
    }
    return "unknown";
}

trade_engine::trade_engine(string name, strategy_engine *strategy) : engine_base(std::move(name)) {
    m_engine_name = "trade";
    m_strategy_engine = strategy;
}

#pragma clang diagnostic pop